package com.kkb.hbase.mr.demo2UDInputFormat;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;

import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import java.io.IOException;

/**
 * 项目：使用自定义InputFormat 合并小文件 为一个大的SequenceFile文件
 * 该自定义输入文件类型类 用于读取k1,v1，类型可以自定义
 * k1: NullWritable, 不需要k1 干任何事情
 * v1：一个文本文件的所有内容 装到字节数组里面 BytesWritable
 */
public class MyInputFormat extends FileInputFormat<NullWritable, BytesWritable> {

    /**
     * 用来读取文件
     * 读取文件的mapTask的个数，取决于文件的切片，【前提文件可以切片】有些压缩文件不能切片，只能使用1个mapTask读取整个文件
     * 把整个文件读过来装到字节数组里面，如果文件可切分，一行一行读，如果文件不可切分，读整个文件内容
     * 文件是否可以切分取决于 isSplitable 方法
     * @param split 整个文件内容（因为不可切分）
     * @param context
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public RecordReader<NullWritable, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        //因为isSplitable 为false ,所以全部读出
        MyRecodReader myRecodReader = new MyRecodReader();
        myRecodReader.initialize(split,context);
        return myRecodReader;
    }

    /**
     * 确定文件是否可以切片
     * @param context
     * @param filename
     * @return
     */
    @Override
    protected boolean isSplitable(JobContext context, Path filename) {
        return false; //表示文件不可以切分，只能 一次性将文件内容全部读出来
    }
}
